package com.rent.message.utils;

import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 批量处理器工厂
 */
@Component
public class BatchProcessorFactory {

    private final Map<String, DistributedBatchProcessor<?>> processors = new ConcurrentHashMap<>();

    /**
     * 创建或获取批量处理器
     */
    @SuppressWarnings("unchecked")
    public <T> DistributedBatchProcessor<T> getOrCreateProcessor(
            String processorName,
            int batchSize,
            long flushInterval,
            Consumer<List<T>> batchConsumer) {

        return (DistributedBatchProcessor<T>) processors.computeIfAbsent(processorName, key -> {
            DistributedBatchProcessor<T> processor = new DistributedBatchProcessor<>(
                    processorName, batchSize, flushInterval, batchConsumer
            );
            processor.start();
            return processor;
        });
    }

    /**
     * 关闭所有处理器
     */
    @PreDestroy
    public void shutdownAll() {
        processors.values().forEach(DistributedBatchProcessor::shutdown);
        processors.clear();
    }
}